package liming.tool.handle.net.udp.socket;

import liming.tool.handle.net.root.RootSocket;
import liming.tool.handle.net.udp.base.UdpBaseMethod;
import liming.tool.handle.net.udp.bean.*;
import liming.tool.pool.*;

import java.io.*;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

/**
 * 作为UDP方向总类，规定基础方法与工具类，方便分化后使用
 * 在udp中，需要考虑数据有效性，与丢包情况，在客户端中，尽可能使用一次连接模式，及向服务器端发送数据后，在收到服务器处理的数据后就关闭连接
 * 在服务器端，采用异步发送与接收，在客户端中，采用同步
 * 核心：发送与接收，数据合并功能，重传机制
 * 发送：将ReceiveMap转为多个NetTemp包并加入到发送队列中（记录ReceiveMap的ID），也可直接将NetTemp包加入到队列中（不记录ID），由发送队列异步发送（不阻塞主线程），并将其放入发送超时缓存队列中等待重发或取消
 * 接收：将收到的DatagramPacket放入receive缓存队列中（不阻塞其他线程）
 */
public abstract class UdpSocket extends RootSocket implements UdpBaseMethod {
    private static final long OVER_TIME = 12_000;//默认超时时间
    private static final float OVER_TIME_RATIO = 0.1f;//默认超限超时比例
    private static final int Reservation = 400;
    protected static int SIZE = 32 * 1024;
    protected int Num;
    private ListenPool<UdpHandleObject1> receivingProcessingPool;//收到的数据,作为接收与初处理的第一处异步
    private ListensPool<UdpHandleObject3> receiveDataMergeProcessingPool;//完整数据处理池，作为数据合并后与上层交付的异步
    private TimeOutPool<UdpHandleObject2> receiveIncompleteDataTimeoutCachePool;//判定超时与未超时的接收数据
    private TimePool<UdpHandleObject4<UdpNetObject>> sendDataPacketCachePool;//发送数据缓存池
    private boolean debug = false;

    /**
     * 默认构造函数，在网络通信中，套接字信息都是一个必要的参数
     *
     * @param datagramSocket Udp监听端口
     */
    public UdpSocket(DatagramSocket datagramSocket, int handleNum) {
        super(datagramSocket.getLocalAddress(), datagramSocket.getLocalPort());
        Num = handleNum;
        debugMessage("创建对象 " + this.getClass().getSimpleName(), new UdpHandleObject(datagramSocket));
        initPools();
    }

    public UdpSocket(InetAddress inetAddress, int port, int handleNum) {
        super(inetAddress, port);
        Num = handleNum;
        initPools();
    }

    /**
     * 将收到的数据包转换为UDP的中间序列化对象
     */
    public static UdpNetObject read(DatagramPacket datagramPacket) throws IOException, ClassNotFoundException {
        ByteArrayInputStream byteStream = new ByteArrayInputStream(datagramPacket.getData());
        ObjectInputStream objectStream = new ObjectInputStream(byteStream);
        return ((UdpNetObject) objectStream.readObject()).setIP(datagramPacket.getAddress(), datagramPacket.getPort());
    }

    /**
     * 将中间对象转为UDPReceiveMap对象
     * （需要完整的中间对象）
     */
    public static UdpReceiveMap read(UdpNetObject udpNetObject) throws IOException, ClassNotFoundException {
        ByteArrayInputStream byteStream = new ByteArrayInputStream(readUdpReceiveMapBytes(udpNetObject.getUdpNetObjects()));
        ObjectInputStream objectStream = new ObjectInputStream(byteStream);
        return ((UdpReceiveMap) objectStream.readObject()).setIP(udpNetObject.getInetAddress(), udpNetObject.getPort());
    }

    /**
     * 将 udpNetObjects中的byte[] 数据读取并拼接
     */
    private static byte[] readUdpReceiveMapBytes(UdpNetObject[] udpNetObjects) throws IOException {
        ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
        for (UdpNetObject udpNetObject : udpNetObjects) {
            byte[] data = udpNetObject.getData();
            outputStream.write(data);
        }
        return outputStream.toByteArray();
    }

    /**
     * 从指定端口发送udp数据集
     *
     * @param datagramSocket 发送端
     * @param inetAddress    接收端套接字
     * @param port           接收端端口
     * @param udpReceiveMap  数据
     */
    public static List<UdpNetObject> write(DatagramSocket datagramSocket, InetAddress inetAddress, int port, UdpReceiveMap udpReceiveMap) throws IOException, ClassNotFoundException {
        UdpReceiveMap UdpReceiveMap_copy = udpReceiveMap.copy();//拷贝一份数据，防止多线程情况下对ID的修改
        List<UdpNetObject> udpNetObjects = write(UdpReceiveMap_copy.getID(), UdpReceiveMap_copy);
        for (UdpNetObject udpNetObject : udpNetObjects) {
            write(datagramSocket, inetAddress, port, udpNetObject);
        }
        return udpNetObjects;
    }

    /**
     * 从指定端口发送udp中间对象包
     */
    protected static void write(DatagramSocket datagramSocket, InetAddress inetAddress, int port, UdpNetObject udpNetObject) throws IOException {
        DatagramPacket datagramPacket = write(udpNetObject, inetAddress, port);
        datagramSocket.send(datagramPacket);
    }

    /**
     * 给定id，给定对象，将其转为UDP中间对象数组
     */
    private static List<UdpNetObject> write(String id, Object object) throws IOException {
        return write(id, write(object));
    }

    /**
     * 将待发送的中间对象天然UDP数据包中
     */
    private static DatagramPacket write(UdpNetObject udpNetObject, InetAddress inetAddress, int port) throws IOException {
        byte[] bytes = write(udpNetObject);
        return new DatagramPacket(bytes, bytes.length, inetAddress, port);
    }

    /**
     * 给定id，给定序列化后的对象，将其转为UDP中间对象数组
     */
    private static List<UdpNetObject> write(String id, byte[] bytes) {
        List<UdpNetObject> udpNetObjects = new ArrayList<>();
        List<UdpNetTempObject> udpNetTempObjects = new ArrayList<>();
        try {
            int length = bytes.length;
            int startIdx = 0, endIdx;
            while (startIdx < length) {
                endIdx = Math.min(startIdx + SIZE - Reservation, length);
                udpNetTempObjects.add(new UdpNetTempObject(id, Arrays.copyOfRange(bytes, startIdx, endIdx)));
                startIdx = endIdx;
            }
            for (int i = 0; i < udpNetTempObjects.size(); i++) {
                udpNetObjects.add(udpNetTempObjects.get(i).getUdpNetObject(i, udpNetTempObjects.size()));
            }
            return udpNetObjects;
        } finally {
            udpNetTempObjects.clear();
        }
    }

    /**
     * 将对象序列化
     */
    private static byte[] write(Object object) throws IOException {
        ByteArrayOutputStream byteStream = new ByteArrayOutputStream();
        ObjectOutputStream objectStream = new ObjectOutputStream(byteStream);
        try {
            objectStream.writeObject(object);
            objectStream.flush();
            return byteStream.toByteArray();
        } finally {
            objectStream.close();
        }
    }

    public static void main(String[] args) throws IOException, ClassNotFoundException {
//        DatagramSocket datagramSocket=new DatagramSocket(6465);
        UdpReceiveMap udpReceiveMap = new UdpReceiveMap();
        udpReceiveMap.put("bytes", new byte[100000]);
        udpReceiveMap.put("123", "123");
        String id = udpReceiveMap.getID();
//        List<UdpNetObject> write = write(datagramSocket, InetAddress.getByName("127.0.0.1"), 6465, udpReceiveMap);
//        List<UdpNetObject> write = write(udpReceiveMap.getID(), udpReceiveMap);
        byte[] bytes = write(udpReceiveMap);
        System.out.println(udpReceiveMap);
        List<UdpNetObject> write = write(id, bytes);
        System.out.println(SIZE + " " + bytes.length);
        System.out.println(read(write.get(0)));
        System.out.println(write.size());
    }

    private void initPools() {
        receivingProcessingPool = new ListenPool<UdpHandleObject1>(Pools.getListPool("System.udp.server@" + hashCode()), 0, OVER_TIME) {
            /**
             * 处理收到的DatagramPacket包，将其转换为内部对象。
             * 前处理方法handleFrontPacketData 在收到数据后将其
             * 后处理方法handleAfterPacketData
             */
            @Override
            public void handle(UdpHandleObject1 udpHandleObject1) throws Exception {
                if (!handleFrontPacketData(udpHandleObject1.getDatagramPacket(), udpHandleObject1.getDatagramSocket())) {
                    return;
                }
                UdpNetObject udpNetObject = read(udpHandleObject1.getDatagramPacket());
                if (!handleAfterPacketData(udpHandleObject1.getDatagramPacket(), udpHandleObject1.getDatagramSocket(), udpNetObject)) {
                    return;
                }
                if (udpNetObject.isValid()) {
                    UdpReceiveMap udpReceiveMap = read(udpNetObject);
                    udpReceiveMap.setLength(udpNetObject.getLength());
                    UdpSocket.this.put(udpReceiveMap, udpHandleObject1.getDatagramSocket());
                } else {
                    handleDataReceiveListenPool(udpNetObject, udpHandleObject1.getDatagramSocket());
                }
            }
        };
        receiveDataMergeProcessingPool = new ListensPool<UdpHandleObject3>(Pools.getListPool("System.udp.temp.UdpHandleObject3@" + hashCode()), 0, OVER_TIME, Num) {
            @Override
            public void handle(UdpHandleObject3 udpHandleObject3) throws Exception {
                handleDataSubmissionPool(udpHandleObject3.getUdpReceiveMap(), udpHandleObject3.getDatagramSocket());
            }
        };
        receiveIncompleteDataTimeoutCachePool = new TimeOutPool<UdpHandleObject2>("System.TimePool.receive@" + hashCode(), OVER_TIME, OVER_TIME_RATIO) {
            @Override
            public void onTimeout(String key, UdpHandleObject2 value) {
                handleReceivePoolOnTimeout(key, value);
            }

            @Override
            public void onRestrict(String key, UdpHandleObject2 value) {
                handleReceivePoolOnRestrict(key, value);
            }
        };
        sendDataPacketCachePool = new TimePool<UdpHandleObject4<UdpNetObject>>("System.TimePool.receive@" + hashCode(), OVER_TIME, 2000) {
            @Override
            public void handle(String key, DataObject<UdpHandleObject4<UdpNetObject>> value) {
                UdpHandleObject4<UdpNetObject> udpHandleObject4 = value.getValue();
                handleSendTimePool(udpHandleObject4.getDatagramSocket(), udpHandleObject4.getInetAddress(), udpHandleObject4.getPort(), key, udpHandleObject4.getList());
            }
        };
    }

    /**
     * 不完整数据处理方法
     *
     * @return 是否放入到接收数据缓冲池中
     */
    @Override
    public UdpNetObject handleDataReceiveListenPool(UdpNetObject udpNetObject, DatagramSocket datagramSocket) throws Exception {
        String id = udpNetObject.getId();
        UdpHandleObject2 dateTemp;
        UdpNetObject tempUdpNetObject = null;
        if (containsKey4Receive(id)) {
            dateTemp = getValue4Receive(id);
            tempUdpNetObject = dateTemp.getUdpNetObject();
            tempUdpNetObject.add(udpNetObject);
            if (tempUdpNetObject.isValid()) {
                receive4Receive(id);
                UdpSocket.this.put(read(tempUdpNetObject).setLength(tempUdpNetObject.getLength()), datagramSocket);
            } else flush4Receive(id);
        } else {
            UdpSocket.this.put(id, udpNetObject, datagramSocket);
        }
        return tempUdpNetObject;
    }

    /**
     * 实现数据提交上层，主要用于当收到完整的UdpReceiveMap后将数据视情况提交到上层的doWork方法。
     * 业务处理模块，由Server或Client自主完成,当返回值为false时，则表示这个数据为底层逻辑，不需要处理，
     * UdpHandleObject3中含UdpReceiveMap和DatagramSocket
     */
    @Override
    public boolean handleDataSubmissionPool(UdpReceiveMap udpReceiveMap, DatagramSocket datagramSocket) throws Exception {
        debugMessage("收到信息 " + udpReceiveMap.getID() + " State:" + udpReceiveMap.isState() + " Rm:" + udpReceiveMap.isRm() + " All:" + udpReceiveMap.isResAll() + " id:" + udpReceiveMap.optString("ID", null));
        if (udpReceiveMap.isRm()) {
            remove4Send(udpReceiveMap.getString("ID"));
            debugMessage("正在删除 " + udpReceiveMap.getID() + " State:" + udpReceiveMap.isState() + " Rm:" + udpReceiveMap.isRm() + " All:" + udpReceiveMap.isResAll() + " id:" + udpReceiveMap.optString("ID", null));
        } else {
            {
                UdpReceiveMap rm = new UdpReceiveMap();
                rm.put("ID", udpReceiveMap.getID());
                InetAddress inetAddress = udpReceiveMap.getInetAddress();//获取套接字
                int port = udpReceiveMap.getPort();//获取端口号
                rm.setRm(true);
                write(datagramSocket, inetAddress, port, rm);
                debugMessage("请求清除 " + rm.getID() + " State:" + rm.isState() + " Rm:" + rm.isRm() + " All:" + rm.isResAll() + " id:" + rm.optString("ID"));
            }
            if (udpReceiveMap.isState()) {
                try {
                    String id = udpReceiveMap.optString("ID");
                    byte[] deletion = udpReceiveMap.optString("deletion").getBytes();
                    UdpHandleObject4<UdpNetObject> value = getValue4Send(id);//获取存储的值
                    List<UdpNetObject> list = value.getList();//获取缓存数据
                    InetAddress inetAddress = value.getInetAddress();//获取套接字
                    int port = value.getPort();//获取端口号
                    List<Integer> deletion1 = UdpNetObject.getDeletion(deletion, list.size());//解析缺失的包
                    flush4Send(id);
                    for (Object object : deletion1) {
                        if (object instanceof Integer) {
                            write(datagramSocket, inetAddress, port, list.get((int) object));
                        }
                    }
                } catch (Exception e) {
                    writeLog(e);
                }
            } else {
                return true;
            }
        }
        return false;
    }

    /**
     * 实现数据接收缓存逻辑之一 ———— 超时逻辑。
     * 当缓存的数据超过规定时间后，就会调用此处方法
     */
    @Override
    public void handleReceivePoolOnTimeout(String key, UdpHandleObject2 value) {
        writeLog(
                "接收已超时 " + value.getUdpNetObject().getId() + " 缺失：" + UdpNetObject.getDeletion(value.getUdpNetObject()));
    }

    /**
     * 实现数据接收缓存逻辑之一 ———— 超限逻辑。
     * 当缓存的数据超过规定时间一定比例后（0~1），就会调用此处方法
     */
    @Override
    public void handleReceivePoolOnRestrict(String key, UdpHandleObject2 value) {
        UdpNetObject udpNetObject = value.getUdpNetObject();
        UdpReceiveMap receiveMap = new UdpReceiveMap();
        receiveMap.setState(true);
        receiveMap.put("deletion", udpNetObject.getValues());
        receiveMap.put("ID", key);
        try {
            send(value.getDatagramSocket(), receiveMap, udpNetObject.getInetAddress(), udpNetObject.getPort());
        } catch (IOException | ClassNotFoundException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 实现数据重新发送缓存逻辑。
     * 当发送的数据超过规定时间后，就会调用此处方法
     */
    public void handleSendTimePool(DatagramSocket datagramSocket, InetAddress inetAddress, int port, String key, List<UdpNetObject> list) {
        put(datagramSocket, list, inetAddress, port, key);
        for (UdpNetObject udpNetObject : list) {
            try {
                write(datagramSocket, inetAddress, port, udpNetObject);
            } catch (IOException e) {
                writeStrongLog("发送失败:" + e.getMessage() + " \t\t" + udpNetObject);
            }
        }
        debugMessage("发送缓存全重发： " + key + " " + list.size() + " " + System.currentTimeMillis());
    }

    /**
     * 设置数据初处理池
     */
    public void setDataReceiveListenPool(ListenPool<UdpHandleObject1> packet2Net) {
        this.receivingProcessingPool.close();
        this.receivingProcessingPool = packet2Net;
    }

    /**
     * 设置数据提交池
     */
    public void setDataSubmissionPool(ListensPool<UdpHandleObject3> net2Map) {
        this.receiveDataMergeProcessingPool.close();
        this.receiveDataMergeProcessingPool = net2Map;
    }

    /**
     * 设置数据接收缓存池
     */
    public void setReceivePool(TimeOutPool<UdpHandleObject2> receive) {
        this.receiveIncompleteDataTimeoutCachePool.stop();
        this.receiveIncompleteDataTimeoutCachePool = receive;
    }

    /**
     * 设置数据发送缓存池
     */
    public void setSendPool(TimePool<UdpHandleObject4<UdpNetObject>> send) {
        this.sendDataPacketCachePool.stop();
        this.sendDataPacketCachePool = send;
    }

    /**
     * 向初处理池添加数据
     */
    protected void put(DatagramPacket datagramPacket, DatagramSocket datagramSocket) {
        receivingProcessingPool.put(new UdpHandleObject1(datagramPacket, datagramSocket));
    }

    protected void put(DatagramSocket datagramSocket, DatagramPacket datagramPacket) {
        receivingProcessingPool.put(new UdpHandleObject1(datagramPacket, datagramSocket));
    }

    /**
     * 向业务提交池添加数据
     */
    protected void put(UdpReceiveMap udpReceiveMap, DatagramSocket datagramSocket) {
        receiveDataMergeProcessingPool.put(new UdpHandleObject3(udpReceiveMap, datagramSocket));
    }

    protected void put(DatagramSocket datagramSocket, UdpReceiveMap udpReceiveMap) {
        receiveDataMergeProcessingPool.put(new UdpHandleObject3(udpReceiveMap, datagramSocket));
    }

    /**
     * 向发送缓存池提交数据
     */
    protected void put(DatagramSocket datagramSocket, List<UdpNetObject> lists, InetAddress inetAddress, int port, String ID) {
        sendDataPacketCachePool.put(ID, new UdpHandleObject4(lists, inetAddress, port, datagramSocket), lists.size() * 10L);
    }

    protected void put(DatagramSocket datagramSocket, InetAddress inetAddress, int port, UdpReceiveMap udpReceiveMap) throws IOException, ClassNotFoundException {
        String ID = udpReceiveMap.getID();
        List<UdpNetObject> lists = write(datagramSocket, inetAddress, port, udpReceiveMap);
        debugMessage("发送ID:" + ID + ",发送包数=" + lists.size() + ",发送目标" + inetAddress + ":" + port);
        put(datagramSocket, lists, inetAddress, port, ID);
    }

    /**
     * 向接收缓存池提交数据
     */
    protected void put(String id, UdpNetObject udpNetObject, DatagramSocket datagramSocket) {
        receiveIncompleteDataTimeoutCachePool.put(id, new UdpHandleObject2(udpNetObject, datagramSocket), udpNetObject.getLength() * 10L);
    }

    /**
     * 刷新发送缓存池指定ID的时间
     */
    protected UdpHandleObject4<UdpNetObject> flush4Send(String ID) {
        return sendDataPacketCachePool.flush(ID);
    }

    /**
     * 得到发送缓存池指定ID的存储的值
     */
    protected UdpHandleObject4<UdpNetObject> getValue4Send(String ID) {
        return sendDataPacketCachePool.getValue(ID);
    }

    protected boolean containsKey4Receive(String ID) {
        return receiveIncompleteDataTimeoutCachePool.containsKey(ID);
    }

    protected UdpHandleObject2 getValue4Receive(String ID) {
        return receiveIncompleteDataTimeoutCachePool.get(ID).getValue();
    }

    protected boolean receive4Receive(String ID) {
        return receiveIncompleteDataTimeoutCachePool.remove(ID) != null;
    }

    protected UdpHandleObject2 flush4Receive(String ID) {
        return receiveIncompleteDataTimeoutCachePool.flush(ID);
    }

    protected UdpHandleObject4<UdpNetObject> remove4Send(String ID) {
        return sendDataPacketCachePool.remove(ID).getValue();
    }

    public void send(DatagramSocket datagramSocket, UdpReceiveMap udpReceiveMap, InetAddress inetAddress, int port) throws IOException, ClassNotFoundException {
        put(datagramSocket, inetAddress, port, udpReceiveMap);
    }

    @Override
    public void close() throws IOException {
        receivingProcessingPool.close();
        receiveDataMergeProcessingPool.close();
        receiveIncompleteDataTimeoutCachePool.stop();
        sendDataPacketCachePool.stop();
    }

    protected void debugMessage(Object... objects) {
        if (debug)
            for (Object object : objects) {
                writeLog("debug: " + object);
            }
    }

    public void setDebug(boolean debug) {
        this.debug = debug;
    }
}
